/*-------------------------------------------------------------------------
 *
 * basebackup_throttle.c
 *	  基本备份接收器实现节流。数据将以不大于
 *	  配置的最大值的速度转发到链中的下一个基本备份接收器。
 *
 * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/backup/basebackup_throttle.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "backup/basebackup_sink.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "storage/latch.h"
#include "utils/timestamp.h"

typedef struct bbsink_throttle
{
	/* 所有类型的接收器的通用信息。 */
	bbsink		base;

	/* 实际的字节数，传输这些字节可能导致休眠。 */
	uint64		throttling_sample;

	/* 已经传输但尚未被限制的数据量。 */
	int64		throttling_counter;

	/* 传输throttling_sample字节所需的最小时间。 */
	TimeOffset	elapsed_min_unit;

	/* 对传输速率的最后检查。 */
	TimestampTz throttled_last;
} bbsink_throttle;

static void fc_bbsink_throttle_begin_backup(bbsink *fc_sink);
static void fc_bbsink_throttle_archive_contents(bbsink *fc_sink, size_t fc_len);
static void fc_bbsink_throttle_manifest_contents(bbsink *fc_sink, size_t fc_len);
static void fc_throttle(bbsink_throttle *fc_sink, size_t fc_increment);

static const bbsink_ops bbsink_throttle_ops = {
	.begin_backup = fc_bbsink_throttle_begin_backup,
	.begin_archive = bbsink_forward_begin_archive,
	.archive_contents = fc_bbsink_throttle_archive_contents,
	.end_archive = bbsink_forward_end_archive,
	.begin_manifest = bbsink_forward_begin_manifest,
	.manifest_contents = fc_bbsink_throttle_manifest_contents,
	.end_manifest = bbsink_forward_end_manifest,
	.end_backup = bbsink_forward_end_backup,
	.cleanup = bbsink_forward_cleanup
};

/*
 * 以指定速率的秒数的分数来决定限制的频率。
 */
#define THROTTLING_FREQUENCY	8

/*
 * 创建一个新的basebackup接收器，该接收器执行限制并将数据转发
 * 到下一个接收器。
 */
bbsink * bbsink_throttle_new(bbsink *fc_next, uint32 fc_maxrate)
{
	bbsink_throttle *fc_sink;

	Assert(fc_next != NULL);
	Assert(fc_maxrate > 0);

	fc_sink = palloc0(sizeof(bbsink_throttle));
	*((const bbsink_ops **) &fc_sink->base.bbs_ops) = &bbsink_throttle_ops;
	fc_sink->base.bbs_next = fc_next;

	fc_sink->throttling_sample =
		(int64) fc_maxrate * (int64) 1024 / THROTTLING_FREQUENCY;

	/*
	 * 传输throttling_sample字节所需的最少时间。
	 */
	fc_sink->elapsed_min_unit = USECS_PER_SEC / THROTTLING_FREQUENCY;

	return &fc_sink->base;
}

/*
 * 这里没有实际的工作要做，但我们需要记录当前时间，以便
 * 用于将来的计算。
 */
static void fc_bbsink_throttle_begin_backup(bbsink *fc_sink)
{
	bbsink_throttle *fc_mysink = (bbsink_throttle *) fc_sink;

	bbsink_forward_begin_backup(fc_sink);

	/* “真实数据”现在开始（头部被忽略）。 */
	fc_mysink->throttled_last = GetCurrentTimestamp();
}

/*
 * 首先进行限制，然后将归档内容传递给下一个接收器。
 */
static void fc_bbsink_throttle_archive_contents(bbsink *fc_sink, size_t fc_len)
{
	fc_throttle((bbsink_throttle *) fc_sink, fc_len);

	bbsink_forward_archive_contents(fc_sink, fc_len);
}

/*
 * 首先进行限制，然后将清单内容传递给下一个接收器。
 */
static void fc_bbsink_throttle_manifest_contents(bbsink *fc_sink, size_t fc_len)
{
	fc_throttle((bbsink_throttle *) fc_sink, fc_len);

	bbsink_forward_manifest_contents(fc_sink, fc_len);
}

/*
 * 将网络传输计数器增加给定的字节数，
 * 如有必要，休眠以符合请求的网络传输
 * 速率。
 */
static void fc_throttle(bbsink_throttle *fc_sink, size_t fc_increment)
{
	TimeOffset	fc_elapsed_min;

	Assert(fc_sink->throttling_counter >= 0);

	fc_sink->throttling_counter += fc_increment;
	if (fc_sink->throttling_counter < fc_sink->throttling_sample)
		return;

	/* 至少应该经过多长时间？ */
	fc_elapsed_min = fc_sink->elapsed_min_unit *
		(fc_sink->throttling_counter / fc_sink->throttling_sample);

	/*
	 * 由于由于并发的WAL活动，闩锁可能会被反复设置，因此
	 * 在循环中休眠以确保足够的时间已经过去。
	 */
	for (;;)
	{
		TimeOffset	fc_elapsed,
					fc_sleep;
		int			fc_wait_result;

		/* 自上次测量以来经过的时间（以及可能的唤醒）。 */
		fc_elapsed = GetCurrentTimestamp() - fc_sink->throttled_last;

		/* 如果传输速度超过预期，进行休眠 */
		fc_sleep = fc_elapsed_min - fc_elapsed;
		if (fc_sleep <= 0)
			break;

		ResetLatch(MyLatch);

		/* 我们正在处理一个潜在设置的闩锁，因此检查中断 */
		CHECK_FOR_INTERRUPTS();

		/*
		 * (TAR_SEND_SIZE / throttling_sample * elapsed_min_unit) 应该是
		 * 最大的休眠时间。因此转换为long是安全的。
		 */
		fc_wait_result = WaitLatch(MyLatch,
								WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
								(long) (fc_sleep / 1000),
								WAIT_EVENT_BASE_BACKUP_THROTTLE);

		if (fc_wait_result & WL_LATCH_SET)
			CHECK_FOR_INTERRUPTS();

		/* 等待完成了吗？ */
		if (fc_wait_result & WL_TIMEOUT)
			break;
	}

	/*
	 * 由于我们使用整数，只有throttling_sample的完整倍数被
	 * 处理。其余部分将在下一次调用此函数时完成。
	 */
	fc_sink->throttling_counter %= fc_sink->throttling_sample;

	/*
	 * 剩余量和可能的下一个增量的时间间隔
	 * 现在开始。
	 */
	fc_sink->throttled_last = GetCurrentTimestamp();
}
